package com.zhaorujie.rpc.core.remoting.netty.server;

import com.zhaorujie.rpc.core.remoting.params.Beat;
import com.zhaorujie.rpc.core.remoting.params.RpcRequest;
import com.zhaorujie.rpc.core.remoting.params.RpcResponse;
import com.zhaorujie.rpc.core.remoting.provider.RpcProviderFactory;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author 赵汝杰
 * @version v1.0
 * @description NettyServerHandler
 * @date 2024/3/5
 **/
public class NettyServerHandler extends SimpleChannelInboundHandler<RpcRequest> {

    private static final Logger log = LoggerFactory.getLogger(NettyServerHandler.class);

    private final ThreadPoolExecutor serviceThreadPool;
    private final RpcProviderFactory rpcProviderFactory;

    public NettyServerHandler(RpcProviderFactory rpcProviderFactory, ThreadPoolExecutor serviceThreadPool) {
        this.rpcProviderFactory = rpcProviderFactory;
        this.serviceThreadPool = serviceThreadPool;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest rpcRequest) throws Exception {
        // 如果是心跳信息则不需要处理
        if (Objects.equals(Beat.BEAR_REQUEST_ID, rpcRequest.getRequestId())) {
            log.debug("rpc provider netty server read beat-ping.");
            return;
        }

        // 自定义线程池处理耗时的业务
        serviceThreadPool.execute(() -> {
            try {
                RpcResponse rpcResponse = rpcProviderFactory.invokeService(rpcRequest);
                log.debug("rpc 响应结果 {}", rpcResponse);
                ctx.writeAndFlush(rpcResponse);
            } catch (Exception e) {
                // 异常处理
                RpcResponse rpcResponse = new RpcResponse();
                rpcResponse.setRequestId(rpcRequest.getRequestId());
                rpcResponse.setThrowable(e);
                ctx.writeAndFlush(rpcResponse);
            }
        });
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        // 心跳判断
        if (evt instanceof IdleStateEvent) {
            ctx.close();
            log.debug("rpc provider netty server close an idle channel.");
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("rpc provider netty server caught exception", cause);
        ctx.close();
    }
}
